package com.im.consumer;
import com.alibaba.fastjson2.JSON;
import com.im.consumer.message.Message;
import com.im.entity.vo.MessageStrategyVO;
import com.im.grpc.ChatMessageProto;
import com.im.grpc.GrpcClientService;
import com.im.grpc.chat.PrivateChatMessage;
import com.im.service.IImPrivateMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class KafkaPushConsumer {
    @Autowired
    private GrpcClientService grpcClientService;
    @Autowired
    private IImPrivateMessageService iImPrivateMessageService;
    /**
     * 监听 Kafka 消息
     * @param message 消息内容
     */
    @KafkaListener(topics = "im-upstream-topic", groupId = "im-group")
    public void consumeMessage(Message message) {
        // 处理消息
        if (message != null) {
            PrivateChatMessage bodyData = JSON.parseObject(message.getBody().getData(), PrivateChatMessage.class);
            log.info("解析后的消息内容：{}", bodyData);
            //首先先走一次消息策略判断的逻辑 IM
            MessageStrategyVO messageStrategy=iImPrivateMessageService.messageStrategy(bodyData);
            if (messageStrategy.isFlag()){
                //消息保存+推送
                PrivateChatMessage privateChatMessage = iImPrivateMessageService.saveMessageAndSessionUpdateAndSendMessage(bodyData);
                //组装一下chat发送(后续删除或kafka目前只先更新会话)
                ChatMessageProto.ChatMessage chatMessage = ChatMessageProto.ChatMessage.newBuilder()
                        .setSessionId(privateChatMessage.getSessionId())
                        .setSenderId(privateChatMessage.getSenderId())
                        .setReceiverId(privateChatMessage.getReceiverId())
                        .setMessageType(privateChatMessage.getMessageType())
                        .setMessageId(privateChatMessage.getMessageId())
                        .setTimestamp(privateChatMessage.getTimestamp())
                        .setContext(privateChatMessage.getContext())
                        .setChatType(ChatMessageProto.ChatMessage.ChatType.SINGLE_CHAT)
                        .build();
                ChatMessageProto.SendMessageResponse response = grpcClientService.sendMessage(chatMessage);// 调用 gRPC
              }else{
                //todo 推送回去消息策略失败
                log.info("消息策略失败：{}",messageStrategy.getMsg());
            }
        } else log.warn("❌ 收到的消息为空！");
    }
}
